package com.nokia.pc3.core.format.input;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import com.nokia.pc3.core.dac.DAC;
import com.nokia.pc3.core.dac.DACInit;
import com.nokia.pc3.core.dac.NokiaUmem;
import com.nokia.pc3.core.xform.Transform;
import com.nokia.pc3.core.xform.TransformFactory;
import com.nokia.pc3.core.xform.Transformer;

public class XFormDriver {

	public static class Map extends
			Mapper<LongWritable, NokiaXFormRecord, NullWritable, Text> {
		private Transform transform = null;
		
		public void map(LongWritable key, NokiaXFormRecord value,
				Context context) throws IOException, InterruptedException {

			NokiaUmem uMem = value.getUmem();
			transform.reset();
			transform.setTransformUmem(uMem);

			while (transform.hasNext()) {
				transform.next();
				Transformer f = transform.getTransformerInstance();
				f.run();
			}

			// Map uMem transformer outputs to oMem as defined by xformOutlist.
			uMem.setLabelOutput(transform.getXformation());

			context.write(NullWritable.get(), new Text(value.toString()));
		}

		public void setup(Context context) {
			DAC dac = new DACInit().getDAC();

			TransformFactory factory = new TransformFactory(dac);
			try {
				transform = factory.getTransform("job");
			} catch (ClassNotFoundException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();

		conf.set("pc3.core.DACInit", "com.nokia.pc3.core.dac.DACInit");

		Job job = new Job(conf);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);

		job.setMapperClass(Map.class);
		job.setInputFormatClass(NokiaUnstructuredRecordInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		job.waitForCompletion(true);
	}

}